package com.example.bookservice.kafka;

import com.example.bookservice.dsl.BookDSL;
import com.example.bookservice.dsl.ChapterDSL;
import com.example.bookservice.entity.Book;
import com.example.bookservice.entity.Chapter;
import com.example.bookservice.service.BookService;
import com.example.bookservice.service.ChapterService;
import com.example.commom.enums.StatusCodeEnum;
import com.example.commom.exception.CustomException;
import com.example.commom.global.KafkaConstants;
import com.example.commom.es.ElasticsearchUpdateCommand;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.transaction.annotation.Transactional;

/**
 * 同步更新或插入ES
 */
@Slf4j
@Configuration
@Transactional
public class KafkaUpdateES {

    @Autowired
    private RestHighLevelClient client;

    @Autowired
    private BookService bookService;

    @Autowired
    private ChapterService chapterService;

    @Autowired
    private ObjectMapper objectMapper;

    @KafkaListener(topics = KafkaConstants.KAFKA_BOOK_UPDATE_INSERT_ES, groupId = KafkaConstants.KAFKA_GROUP_ID)
    public void insertOrUpdateBookES(String bookId, Acknowledgment acknowledgment) {

        log.info("收到消息: " + bookId);
        // 从数据库查询书籍信息
        Book book = bookService.getById(bookId);
        if (book != null) {
            try {
                // 0. 转换为文档类型
                BookDSL bookDSL = new BookDSL(book);

                // 1. 准备 request
                IndexRequest request = new IndexRequest("book").id(bookId);

                // 准备 DSL 语句
                request.source(objectMapper.writeValueAsString(bookDSL), XContentType.JSON);
                // 准备 Hystrix
                ElasticsearchUpdateCommand command = new ElasticsearchUpdateCommand(client, request);
                // 执行Hystrix命令, 发起请求
                boolean success = command.execute();
                System.out.println(success);
                if (success) {
                    acknowledgment.acknowledge();
                } else {
                    throw new CustomException(StatusCodeEnum.FAILED.getMsg(), StatusCodeEnum.FAILED.getCode());
                }
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }

        } else {
            acknowledgment.acknowledge();
        }

    }


    @KafkaListener(topics = KafkaConstants.KAFKA_CHAPTER_INSERT_UPDATE_ES, groupId = KafkaConstants.KAFKA_GROUP_ID)
    public void insertOrUpdateChapterES(String chapterId, Acknowledgment acknowledgment) {

        log.info("收到消息: " + chapterId);
        // 从数据库查询书籍信息
        Chapter chapter = chapterService.getById(chapterId);
        if (chapter != null) {
            try {
                // 0. 转换为文档类型
                ChapterDSL chapterDSL = new ChapterDSL(chapter);

                // 1. 准备 request
                IndexRequest request = new IndexRequest("chapter").id(chapterId);

                // 准备 DSL 语句
                request.source(objectMapper.writeValueAsString(chapterDSL), XContentType.JSON);
                // 准备 Hystrix
                ElasticsearchUpdateCommand command = new ElasticsearchUpdateCommand(client, request);
                // 执行Hystrix命令, 发起请求
                boolean success = command.execute();
                if (success) {
                    acknowledgment.acknowledge();
                } else {
                    throw new CustomException(StatusCodeEnum.FAILED.getMsg(), StatusCodeEnum.FAILED.getCode());
                }
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }

        } else {
            acknowledgment.acknowledge();
        }

    }

}
